package com.ds.lens.flink.monitor.monitorjob.function.detector;

import com.ds.lens.data.common.constant.AggregatorEnum;
import com.ds.lens.data.common.constant.CompareOperatorEnum;
import com.ds.lens.data.common.util.Pair;
import org.apache.flink.api.java.tuple.Tuple2;
import org.springframework.expression.Expression;
import org.springframework.expression.ExpressionParser;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.expression.spel.support.StandardEvaluationContext;

import java.util.HashSet;
import java.util.Set;

/**
 * describe:
 *
 * @author wangyiming
 * @date 2019/05/17
 */
public class RuleExpressionUtil {
    public static final String LOGIC_SPLITER = "(?:\\s+and\\s+|\\s+or\\s+)";
    public static final String COMPARE_SPLITER = "\\b(?:\\s*>\\s*|\\s*>=\\s*|\\s*<\\s*|\\s*<=\\s*|\\s*==\\s*|\\s*!=\\s*)\\b";
    public static final String METRIC_SPLITER = "_";
    private static final String EXPRESSION_PREFIX = "#";
    private static final String METRIC_NAME_PREFIX = "LENS_";

    public static Set<String> getParameterNames(String expression) {
        Set<String> set = new HashSet<>();
        String[] statements = expression.split(LOGIC_SPLITER);
        for (String stat : statements) {
            String[] operators = stat.split(COMPARE_SPLITER);
            set.add(operators[0].substring(1));
        }
        return set;
    }

    public static boolean evaluateExpression(String expression, Set<Tuple2<String, Double>> parameters) {
        ExpressionParser expressionParser = new SpelExpressionParser();
        Expression expressionObj = expressionParser.parseExpression(expression);
        StandardEvaluationContext context = new StandardEvaluationContext();
        for (Tuple2<String, Double> param : parameters) {
            context.setVariable(param.f0, param.f1);
        }
        return expressionObj.getValue(context, Boolean.class);
    }

    public static String composeMetricName(String metricName, AggregatorEnum aggrEnum) {
        if (aggrEnum == null) {
            return metricName;
        }
        return METRIC_NAME_PREFIX + metricName + METRIC_SPLITER + aggrEnum.name();
    }

    public static String composeMetricName(String metricName, Integer aggrNameValue) {
        AggregatorEnum aggrEnum = AggregatorEnum.valueOf(aggrNameValue);
        return METRIC_NAME_PREFIX + metricName + METRIC_SPLITER + aggrEnum.name();
    }

    public static Pair<String, AggregatorEnum> decomposeMetricName(String metricName) {
        if (metricName.startsWith(METRIC_NAME_PREFIX)) {
            metricName = metricName.substring(METRIC_NAME_PREFIX.length());
        }
        int lastIndex = metricName.lastIndexOf(METRIC_SPLITER);
        if (lastIndex == -1 || lastIndex == 0 || lastIndex == metricName.length() - 1) {
            return new Pair<>(metricName, null);
        }
        String aggr = metricName.substring(lastIndex + 1);
        AggregatorEnum aggrEnum = null;
        try {
            aggrEnum = AggregatorEnum.valueOf(aggr);
        } catch (IllegalArgumentException e) {
        }
        return new Pair<>(metricName, aggrEnum);
    }

    public static String buildCompareExpression(String metricName, CompareOperatorEnum operatorEnum, double threshold) {
        return EXPRESSION_PREFIX + metricName + operatorEnum.getOperator() + threshold;
    }
}
